package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 通过这个类 来实现消费消息
 */
public class ConsumerManger {
    //持有上层的 VirtualHost 对象的引用 用来操作数据
    private VirtualHost parent;

    //指定一个线程池 负责去执行具体的回调任务
    private ExecutorService workerPool= Executors.newFixedThreadPool(4);

    //存放令牌的队列 当有队列收到消息的时候 就加入到这个队列 当作令牌
    private BlockingQueue<String> tokenQueue=new LinkedBlockingDeque<>();

    //扫描线程
    private Thread scannerThread=null;

    public ConsumerManger(VirtualHost p){
        parent=p;
        //扫描线程 一直不停的 取令牌队列中的 令牌 也就是有消息的队列
        //并让消费者消费消息
        scannerThread=new Thread(()->{
            while (true){
                try {
                    //取出有消息的队列名(令牌)
                    String queueName = tokenQueue.take();
                    //根据令牌 找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[ConsumerManger] 取令牌后发现 队列名不存在! queueName="+queueName);
                    }
                    //让消费者消费队列中的消息 也就是取出队列中的消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        //线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //将令牌加入循环队列 当队列收到消息时执行
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    //订阅队列的消息 实际上就是将 消费者对象 添加到队列中
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MemoryDataCenter memoryDataCenter = parent.getMemoryDataCenter();
        MSGQueue queue=memoryDataCenter.getQueue(queueName);
        if(queue==null){
            throw new MqException("[ConsumerManger] 队列不存在 queueName="+queueName);
        }
        //构造消费者对象
        ConsumerEnv consumerEnv=new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            //如果当前队列中已经有了一些消息了 需要立即被这个消费者消费掉
            int messageCount = memoryDataCenter.getMessageCount(queueName);
            for (int i = 0; i <messageCount; i++) {
                //这个方法是 调用一次 就消费一条消息
                consumeMessage(queue);
            }
        }

    }

    private void consumeMessage(MSGQueue queue) {
        //1.按照轮询方式 找个消费者 让消费者消费这个消息
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(luckyDog==null){
            //当前队列没有消费者 暂时不消费
            return;
        }
        //2.从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message==null){
            //当前队列中没有消息 无需消费
            return;
        }
        //3.把消息带入到消费者的回调方法中 丢给线程池去执行
        workerPool.submit(()->{
            try {
                //1.把消息放到待确认的集合之中  需要在执行回调方法之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2.执行回调方法
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //3.如果当前是自动应答 此时就可以删除消息
                //如果是手动应答 则 先不处理 交给后续消费调用 basicAck 方法来处理
                if (luckyDog.isAutoAck()) {
                    //1.删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                        //3.删除内存中 消息中心的 消息
                        parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                        //2.删除待确认集合中的 消息
                        parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                        System.out.println("[ConsumerManage] 消息被成功消费 queueName=" + queue.getName());
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }






}
